package com.small.nacos.client.config.impl;


import com.alibaba.nacos.common.http.HttpRestResult;
import small.common.core.constants.CommonConstants;
import small.common.core.constants.ConfigType;
import small.common.crypto.util.KeyUtil;
import small.common.core.util.StringUtil;
import com.small.nacos.api.common.ApiConstants;
import com.small.nacos.api.config.common.LongPullingConstants;
import com.small.nacos.api.config.listener.Listener;
import com.small.nacos.api.exception.NacosException;
import com.small.nacos.client.config.filter.impl.ConfigResponse;
import com.small.nacos.client.config.http.HttpAgent;
import com.small.nacos.client.utils.GroupUtils;
import com.small.nacos.client.utils.ParamUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.HttpURLConnection;
import java.net.URLDecoder;
import java.util.*;
import java.util.concurrent.*;


/**
 * @Author zhoujin
 * @Date 2022/11/18 17:25
 */
public class ClientWorker {

    private final static Logger LOGGER = LoggerFactory.getLogger(ClientWorker.class);

    private final HttpAgent agent;

    private final Properties properties;

    private final ScheduledExecutorService executor;

    private final ScheduledExecutorService executorService;

    //当前长轮询任务
    private int currentLongingTaskCount = 0;

    //超时时间
    private long timeout = 30000L;

    //任务重试事件
    private int taskPenaltyTime = 2000;

    private final ConcurrentHashMap<String, CacheData> cacheMap = new ConcurrentHashMap();

    public ClientWorker(HttpAgent agent, Properties properties) {
        this.agent = agent;
        this.properties = properties;
        this.executor = Executors.newScheduledThreadPool(1, new ThreadFactory() {
            @Override
            public Thread newThread(Runnable runnable) {
                Thread thread = new Thread(runnable);
                thread.setDaemon(true);
                thread.setName("com.alibaba.nacos.client.Worker." + agent.getName());
                return thread;
            }
        });
        this.executorService = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors(), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable runnable) {
                Thread thread = new Thread(runnable);
                thread.setDaemon(true);
                thread.setName("com.alibaba.nacos.client.Worker.longPolling." + agent.getName());
                return thread;
            }
        });
        executor.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                try {
                    checkConfigInfo();
                } catch (Throwable e) {
                    LOGGER.error("[" + agent.getName() + "] [sub-check] rotate check error", e);
                }

            }
        }, 1L, 10L, TimeUnit.MILLISECONDS);
    }

    public ConfigResponse getServerConfig(String dataId, String group, String tenant, long readTimeout) throws NacosException {
        ConfigResponse configResponse = new ConfigResponse();
        if (StringUtil.isBlank(group)) {
            group = ApiConstants.DEFAULT_GROUP;
        }

        HttpRestResult<String> result = null;
        try {
            Map<String, String> params = new HashMap<String, String>(3);
            if (StringUtil.isBlank(tenant)) {
                params.put("dataId", dataId);
                params.put("group", group);
            } else {
                params.put("dataId", dataId);
                params.put("group", group);
                params.put("tenant", tenant);
            }
            result = agent.httpGet(ApiConstants.CONFIG_CONTROLLER_PATH, null, params, agent.getEncode(), readTimeout);
        } catch (Exception ex) {
            String message = String
                    .format("[%s] [sub-server] get server config exception, dataId=%s, group=%s, tenant=%s",
                            agent.getName(), dataId, group, tenant);
            LOGGER.error(message, ex);
            throw new NacosException(NacosException.SERVER_ERROR, ex);
        }

        switch (result.getCode()) {
            //请求成功
            case HttpURLConnection.HTTP_OK:
                //保存配置文件内容
                LocalConfigInfoProcessor.saveSnapshot(agent.getName(), dataId, group, tenant, result.getData());
                configResponse.setContent(result.getData());
                //设置配置文件格式
                String configType;
                if (result.getHeader().getValue(ApiConstants.CONFIG_TYPE) != null) {
                    configType = result.getHeader().getValue(ApiConstants.CONFIG_TYPE);
                } else {
                    configType = ConfigType.TEXT.getType();
                }
                configResponse.setConfigType(configType);
                return configResponse;
            //没有找到配置类
            case HttpURLConnection.HTTP_NOT_FOUND:
                LocalConfigInfoProcessor.saveSnapshot(agent.getName(), dataId, group, tenant, null);
                return configResponse;
            //连接失败,未获取到锁
            case HttpURLConnection.HTTP_CONFLICT: {
                LOGGER.error(
                        "[{}] [sub-server-error] get server config being modified concurrently, dataId={}, group={}, "
                                + "tenant={}", agent.getName(), dataId, group, tenant);
                throw new NacosException(NacosException.CONFLICT,
                        "data being modified, dataId=" + dataId + ",group=" + group + ",tenant=" + tenant);
            }
            //连接被拒绝
            case HttpURLConnection.HTTP_FORBIDDEN: {
                LOGGER.error("[{}] [sub-server-error] no right, dataId={}, group={}, tenant={}", agent.getName(),
                        dataId, group, tenant);
                throw new NacosException(result.getCode(), result.getMessage());
            }
            //默认错误
            default: {
                LOGGER.error("[{}] [sub-server-error]  dataId={}, group={}, tenant={}, code={}", agent.getName(),
                        dataId, group, tenant, result.getCode());
                throw new NacosException(result.getCode(),
                        "http error, code=" + result.getCode() + ",dataId=" + dataId + ",group=" + group + ",tenant="
                                + tenant);
            }

        }
    }

    public void checkConfigInfo() {
        //这里每有3000个配置就会重新加一个线程执行
        int size = cacheMap.size();
        int longingTaskCount = (int) Math.ceil(size / ParamUtil.getPerTaskConfigSize());
        if (longingTaskCount > currentLongingTaskCount) {
            for (int i = currentLongingTaskCount; i < longingTaskCount; i++) {
                executorService.execute(new LongPollingRunnable(i));
            }
            currentLongingTaskCount = longingTaskCount;
        }

    }

    public void addTenantListeners(String dataId, String group, List<Listener> asList) throws NacosException {
        group = GroupUtils.blank2defaultGroup(group);
        String tenant = agent.getTenant();
        CacheData cacheData = addCacheDataIfAbsent(dataId, group, tenant);
        for (Listener listener : asList) {
            cacheData.addListener(listener);
        }
    }

    /**
     * 添加配置缓存
     *
     * @param dataId
     * @param group
     * @param tenant
     * @return
     * @throws NacosException
     */
    public CacheData addCacheDataIfAbsent(String dataId, String group, String tenant) throws NacosException {
        String key = KeyUtil.getKey(dataId, group, tenant);
        CacheData cacheData = cacheMap.get(key);
        if (cacheData != null) {
            return cacheData;
        }
        cacheData = new CacheData(agent.getName(), dataId, group, tenant);

        CacheData lastCacheData = cacheMap.putIfAbsent(key, cacheData);
        //如果之前没有配置缓存
        if (lastCacheData == null) {
            //设置任务id
            cacheData.setTaskId(cacheMap.size() / (int) ParamUtil.getPerTaskConfigSize());
            lastCacheData = cacheData;
        }
        lastCacheData.setInitializing(true);
        return lastCacheData;
    }


    List<String> checkUpdateDataIds(List<CacheData> cacheDatas, List<String> inInitializingCacheList) throws Exception {
        StringBuilder sb = new StringBuilder();
        for (CacheData cacheData : cacheDatas) {
            if (!cacheData.isUseLocalConfig()) {
                String key = null;
                if (StringUtil.isBlank(cacheData.tenant)) {
                    key = KeyUtil.getKey(cacheData.dataId,cacheData.group);
                } else {
                    key = KeyUtil.getKey(cacheData.dataId,cacheData.group,cacheData.tenant);
                }
                KeyUtil.appendKeyAndValue(key,cacheData.getMd5(),sb);
                if (cacheData.isInitializing()) {
                    // 添加初始化配置类
                    inInitializingCacheList
                            .add(KeyUtil.getKey(cacheData.dataId, cacheData.group, cacheData.tenant));
                }
            }
        }
        boolean isInitializingCacheList = !inInitializingCacheList.isEmpty();
        //查询改变的配置类
        return checkUpdateConfigStr(sb.toString(), isInitializingCacheList);
    }

    List<String> checkUpdateConfigStr(String probeUpdateString, boolean isInitializingCacheList) throws Exception {

        Map<String, String> params = new HashMap<String, String>(2);
        params.put(LongPullingConstants.PROBE_MODIFY_REQUEST, probeUpdateString);
        Map<String, String> headers = new HashMap<String, String>(2);
        //长轮询事件
        headers.put(LongPullingConstants.LONG_PULLING_TIMEOUT, "" + timeout);

        //告诉服务器如果添加了新的初始化缓存数据,不进行长轮询
        if (isInitializingCacheList) {
            headers.put(LongPullingConstants.LONG_PULLING_TIMEOUT_NO_HANGUP, "true");
        }

        if (StringUtil.isBlank(probeUpdateString)) {
            return Collections.emptyList();
        }

        try {
            long readTimeoutMs = timeout + (long) Math.round(timeout >> 1);
            HttpRestResult<String> result = agent
                    .httpPost(ApiConstants.CONFIG_CONTROLLER_PATH + "/listener", headers, params, agent.getEncode(),
                            readTimeoutMs);

            if (result.ok()) {
                return parseUpdateDataIdResponse(result.getData());
            } else {
                LOGGER.error("[{}] [check-update] get changed dataId error, code: {}", agent.getName(),
                        result.getCode());
            }
        } catch (Exception e) {
            LOGGER.error("[" + agent.getName() + "] [check-update] get changed dataId exception", e);
            throw e;
        }
        return Collections.emptyList();
    }

    /**
     * 解析更新的配置
     *
     * @param response
     * @return
     */
    private List<String> parseUpdateDataIdResponse(String response) {
        if (StringUtil.isBlank(response)) {
            return Collections.emptyList();
        }

        try {
            response = URLDecoder.decode(response, "UTF-8");
        } catch (Exception e) {
            LOGGER.error("[" + agent.getName() + "] [polling-resp] decode modifiedDataIdsString error", e);
        }

        List<String> updateList = new LinkedList<String>();

        for (String dataIdAndGroup : response.split(CommonConstants.LINE_SEPARATOR)) {
            if (!StringUtil.isBlank(dataIdAndGroup)) {
                String[] keyArr = dataIdAndGroup.split(CommonConstants.WORD_SEPARATOR);
                String dataId = keyArr[0];
                String group = keyArr[1];
                if (keyArr.length == 2) {
                    updateList.add(KeyUtil.getKey(dataId, group));
                    LOGGER.info("[{}] [polling-resp] config changed. dataId={}, group={}", agent.getName(), dataId,
                            group);
                } else if (keyArr.length == 3) {
                    String tenant = keyArr[2];
                    updateList.add(KeyUtil.getKey(dataId, group, tenant));
                    LOGGER.info("[{}] [polling-resp] config changed. dataId={}, group={}, tenant={}", agent.getName(),
                            dataId, group, tenant);
                } else {
                    LOGGER.error("[{}] [polling-resp] invalid dataIdAndGroup error {}", agent.getName(),
                            dataIdAndGroup);
                }
            }
        }
        return updateList;
    }

    class LongPollingRunnable implements Runnable {
        //任务id
        private final int taskId;
        //超时时间
        private final long readTimeOut = 3000L;

        public LongPollingRunnable(int taskId) {
            this.taskId = taskId;
        }

        @Override
        public void run() {
            //由当前线程处理的任务
            List<CacheData> cacheDatas = new ArrayList<CacheData>();
            //首次初始化的配置集合
            List<String> inInitializingCacheList = new ArrayList<String>();
            try {
                for (CacheData cacheData : cacheMap.values()) {
                    if (taskId == cacheData.getTaskId()) {
                        cacheDatas.add(cacheData);
                    }
                }
                //获取改变的配置
                List<String> changedGroupKeys = checkUpdateDataIds(cacheDatas, inInitializingCacheList);

                //重新获取最新配置
                for (String groupKey : changedGroupKeys) {
                    String[] key = KeyUtil.parseKey(groupKey);
                    String dataId = key[0];
                    String group = key[1];
                    String tenant = null;
                    if (key.length == 3) {
                        tenant = key[2];
                    }
                    try {
                        ConfigResponse response = getServerConfig(dataId, group, tenant, readTimeOut);
                        CacheData cache = cacheMap.get(KeyUtil.getKey(dataId, group, tenant));
                        if (cache != null) {
                            cache.setContent(response.getContent());
                            if (null != response.getConfigType()) {
                                cache.setType(response.getConfigType());
                            }
                            LOGGER.info("[{}] [data-received] dataId={}, group={}, tenant={}, md5={}, content={}, type={}",
                                    agent.getName(), dataId, group, tenant, cache.getMd5(),
                                    response.getContent(), response.getConfigType());
                        }
                    } catch (NacosException ioe) {
                        String message = String
                                .format("[%s] [get-update] get changed config exception. dataId=%s, group=%s, tenant=%s",
                                        agent.getName(), dataId, group, tenant);
                        LOGGER.error(message, ioe);
                    }
                }

                //触发监听器
                for (CacheData cacheData : cacheDatas) {
                    // 排除failover文件
                    if (!cacheData.isInitializing() || inInitializingCacheList
                            .contains(KeyUtil.getKey(cacheData.dataId, cacheData.group, cacheData.tenant))) {
                        cacheData.checkListenerMd5();
                        cacheData.setInitializing(false);
                    }
                }
                inInitializingCacheList.clear();
                //都执行完成以后，再次提交长轮询任务
                executorService.execute(this);

            } catch (Throwable e) {
                //如果长轮询执行发生异常，延迟2s执行下一次长轮询
                LOGGER.error("longPolling error : ", e);
                executorService.schedule(this, taskPenaltyTime, TimeUnit.MILLISECONDS);
            }
        }


    }


}
